package main

import (
	"os"
	"externalsort/pipeline"
	"bufio"
	"log"
)

func main()  {
	p := CreatePipeLine("small.in",800000000,8)
	writeToFile(p,"small.out")
	//printFile("small.out")
}
func CreatePipeLine(
	filename string,fileSize,chunkCount int)  <- chan int {
	chunkSize :=fileSize/chunkCount
	log.Printf("file size is %d,chunkcount is:%d and chunksize is:%d",
		fileSize,chunkCount,chunkSize)
	sortResults := []<-chan int{}
	pipeline.Init()
	for i:=0;i< chunkCount;i++{
		file,err := os.Open("small.in")
		if err != nil {
			panic("file open err")
		}
		pos:=int64(i*chunkSize)
		log.Printf("chunk %d read point is:%d",i,int(pos))
		file.Seek(pos,0)
		source:=pipeline.ReadSource(
			bufio.NewReader(file),chunkSize)
		sortResults=append(sortResults,pipeline.InMemorySort(source))
	}
	log.Printf("len of sortResult is:%d",len(sortResults))
	return pipeline.MergeN(sortResults...)
}
func writeToFile(in <- chan int,fileName string){
	file,err :=os.Create(fileName)
	defer file.Close()
	if err != nil {
		log.Println(err)
		panic("file close err")
	}
	write :=bufio.NewWriter(file)
	defer write.Flush()
	pipeline.WriteSink(write,in)
}
func printFile(fileName string){
	file,err:=os.Open(fileName)
	if err != nil {
		panic("file open errr")
	}
	defer file.Close()
	p:=pipeline.ReadSource(file,-1)
	for v :=range p {
		log.Println(v)
	}
}
